package cn.itcast.tags.ml.rs.rdd

import org.apache.spark.mllib.evaluation.RegressionMetrics
import org.apache.spark.mllib.recommendation.{ALS, MatrixFactorizationModel, Rating}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author: xu
 * @desc: 使用MovieLens 电影评分数据集，调用Spark MLlib中的协同过滤推荐算法ALS建立推荐模型：
 *        1、 预测 用户（User）对某个电影（Product）的评分
 *        2、 为某个用户推荐10个电影（Products）
 *        3、 为某个电影推荐10个用户（Users）
 */
object SparkAlsRmdMovie {
  def main(args: Array[String]): Unit = {
    // TODO: 1. 创建Spark实例对象
    val sc: SparkContext = {
      // a. 创建SparkConf 设置应用信息
      val sparkConf = new SparkConf()
        .setMaster("local[4]")
        .setAppName(this.getClass.getSimpleName.stripSuffix("$"))
      // b. 建造者模式构建SparkSession对象
      val context = SparkContext.getOrCreate(sparkConf)
      // 设置检查点目录
      context.setCheckpointDir(s"datas/ckpt/als-ml-${System.nanoTime()}")
      // c. 返回对象
      context
    }

    // TODO: 2. 读取电影评分数据
    val rawRationsRDD: RDD[String] = sc.textFile("datas/als/ml-100k/u.data")
    println(s"Count = ${rawRationsRDD.count()}")
    println(s"Count = ${rawRationsRDD.first()}")

    // TODO: 3. 数据转换，构建RDD[Rating]
    val ratingsRDD: RDD[Rating] = rawRationsRDD
      .filter(line => line != null && line.split("\\t").length == 4)
      .map { line =>
        // 切割字符串，拆箱
        val Array(userId, movieId, rating, _) = line.split("\\t")
        // 返回mllib包下的Rating示例对象
        Rating(userId.toInt, movieId.toInt, rating.toDouble)
      }
    // 划分数据为 => 训练数据集 和 测试数据集
    val Array(trainRatings, testRatings) = ratingsRDD.randomSplit(Array(0.8, 0.2))

    // TODO: 4. 调用ALS算法中显示训练函数训练模型
    // 迭代次数为20，特征数为10
    val alsModel = ALS.train(
      ratings = trainRatings, // 训练数据集
      rank = 10, // 特征数rank
      iterations = 20 // 迭代次数
    )

    // TODO: 5. 获取模型中两个因子矩阵
    /**
     * 获取模型MatrixFactorizationModel中包含的两个矩阵：
     * a. 用户因子矩阵：alsModel.userFeatures
     * b. 产品因子矩阵：alsModel.productFeatures
     */
    // userId -> Features
    val userFeatures: RDD[(Int, Array[Double])] = alsModel.userFeatures
    userFeatures.take(10).foreach { tuple =>
      println(tuple._1 + " -> " + tuple._2.mkString(","))
    }
    println("=========================================================")
    // productId -> Features
    val productFeatures: RDD[(Int, Array[Double])] = alsModel.productFeatures
    productFeatures.take(10).foreach { tuple =>
      println(tuple._1 + " -> " + tuple._2.mkString(","))
    }

    // TODO: 6.模型评估，使用RMSE评估模型，值越小，表示误差越小，模型越好
    // 6.1 转换测试数据集格式为RDD[((userId, ProductId), rating)]
    val actualRatingsRDD: RDD[((Int, Int), Double)] = testRatings.map { tuple =>
      ((tuple.user, tuple.product), tuple.rating)
    }
    // 6.2 使用模型对测试数据集预测电影评分
    val predictRatingsRDD: RDD[((Int, Int), Double)] = alsModel
      // 根据userId，和productId预测评分
      .predict(actualRatingsRDD.map(_._1))
      // 转换数据格式为RDD[((userId, ProductId), rating)]
      .map(tuple => ((tuple.user, tuple.product), tuple.rating))
    // 6.3 合并预测值和真实值
    val predictAndActualRatingsRDD: RDD[((Int, Int), (Double, Double))] = predictRatingsRDD.join(actualRatingsRDD)
    // 6.4 模型评估，计算RMSE的值
    val metrics = new RegressionMetrics(predictAndActualRatingsRDD.map(_._2))
    println(s"RMSE = ${metrics.rootMeanSquaredError}")

    // TODO: 7.推荐与预测评分
    // 7.1 预测某个用户对某个产品的评分，def predict(user: Int, product: Int): Double
    val predictRating: Double = alsModel.predict(196, 242)
    println(s"预测196号用户对242号电影的评分: ${predictRating}")
    println("==========================================")

    // 7.2 给某个用户推荐十部电影，def recommendProducts(user: Int, num: Int): Array[Rating]
    val recommendTenMovies = alsModel.recommendProducts(196, 10)
    println(s"给196号用户推荐10部电影:")
    recommendTenMovies.foreach(println)
    println("==========================================")

    // 7.3 给某部电影推荐十个用户，def recommendUsers(product: Int, num: Int): Array[Rating]
    val recommendTenUsers: Array[Rating] = alsModel.recommendUsers(242, 10)
    println("给242号电影推荐10个用户: ")
    recommendTenUsers.foreach(println)

    // TODO: 8. 将训练的模型进行保存，以便后期使用
    val saveModelPath = "datas/als/ml-als-model-" + System.nanoTime()
    alsModel.save(sc, saveModelPath)

    // TODO: 9. 从文件系统中加载保存的模型，用于推荐预测
    val loadModelPath = MatrixFactorizationModel.load(sc, saveModelPath)
    // 使用加载到的模型，进行预测
    val loadModelPredictRating = loadModelPath.predict(196, 242)
    println(s"使用加载模型，预测196号用户对242号电影的评分: ${loadModelPredictRating}")

    // 应用结束，关闭资源
    sc.stop()
  }
}
